package cn.itcast.flink.transformation;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.concurrent.TimeUnit;

/**
 * @author lilulu
 */
//富函数RichFunction，生命周期方法：open和close，当算子对数据库操作时，建议使用
public class TransformationRichDemo {

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Student {
        private Integer id;
        private String name;
        private Integer age;
    }

    //    自定义数据源，从MySQL表中加载数据，并且实现增量加载
    public static class MysqlSource extends RichParallelSourceFunction<Student> {
        // 定义变量，标识是否加载数据
        private boolean isRunning = true;

        // 定义变量，open方法初始化，close方法关闭连接
        private Connection conn = null;
        private PreparedStatement pstmt = null;
        private ResultSet result = null;

        @Override
        public void open(Configuration parameters) throws Exception {
            System.out.println("调用open()方法，初始化操作：创建数据库连接=========");
            Class.forName("com.mysql.jdbc.Driver");
            conn = DriverManager.getConnection("jdbc:mysql://node1:3306/?useSSL=false", "root", "123456");
            pstmt = conn.prepareStatement("select id,name,age from db_flink.t_student");

        }

        @Override
        public void run(SourceContext<Student> sourceContext) throws Exception {
            while (isRunning) {
                System.out.println("run方法执行，查询数据库数据");
                result = pstmt.executeQuery();
                while (result.next()) {
                    Student student = new Student(
                            result.getInt("id"),
                            result.getString("name"),
                            result.getInt("age")
                    );
                    sourceContext.collect(student);
                }
//                    每隔5秒加载一次数据
                TimeUnit.SECONDS.sleep(5);
            }
        }

        @Override
        public void cancel() {
            isRunning = false;
        }


        @Override
        public void close() throws Exception {
            System.out.println("调用close()方法，收尾工作，关闭数据库连接.....................");
            if (result != null) {
                result.close();
            }
            if (pstmt != null) {
                pstmt.close();
            }
            if (conn != null) {
                conn.close();
            }
        }
    }

    public static void main(String[] args) throws Exception {
        // 1. 执行环境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 2. 数据源-source
        DataStreamSource<Student> studentDataStreamSource = env.addSource(new MysqlSource());
        // 3. 数据转换-transformation
        // 4. 数据终端-sink
        studentDataStreamSource.print();
        // 5. 触发执行-execute
        env.execute("TransformationRichDemo");
    }
}